package groupOne.sink;

import com.alibaba.druid.pool.DruidDataSource;
import com.alibaba.druid.pool.DruidPooledConnection;
import com.alibaba.fastjson.JSONObject;

import groupOne.bean.TableProcess;
import groupOne.util.DruidDSUtil;
import groupOne.util.RedisUtil;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import redis.clients.jedis.Jedis;

import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.Arrays;
import java.util.List;

public class PhoenixSink extends RichSinkFunction<Tuple2<JSONObject, TableProcess>> {

    private DruidDataSource druidDataSource;
    private Jedis redisClient;

    @Override
    public void open(Configuration parameters) throws Exception {
        druidDataSource = DruidDSUtil.getDruidDataSource();
        redisClient = RedisUtil.getRedisClient();
    }

    @Override
    public void close() throws Exception {
        druidDataSource.close();
        redisClient.close();
    }

    @Override
    public void invoke(Tuple2<JSONObject, TableProcess> value, Context context) throws Exception {
        toPhoenix(value);

        delRedis(value);

    }

    private void delRedis(Tuple2<JSONObject, TableProcess> value) {
        JSONObject data = value.f0;
        TableProcess tp = value.f1;
        String key = tp.getSinkTable()+":"+data.getString("id");
        String opType = data.getString("op_type");
        if ("update".equals(opType)) {
            redisClient.del(key);
        }
    }

    private void toPhoenix(Tuple2<JSONObject, TableProcess> value) throws SQLException {
        DruidPooledConnection conn = druidDataSource.getConnection();
        JSONObject data = value.f0;
        TableProcess tp = value.f1;
        // 1. 拼接sql语句. 一定有占位符
        // upsert into user(aa,b,c)values(?,?,?)
        StringBuilder sql = new StringBuilder()
                .append("upsert into ")
                .append(tp.getSinkTable())
                .append("(")
                .append(tp.getSinkColumns())
                .append(")values(")
                .append(tp.getSinkColumns().replaceAll("[^,]+","?"))
                .append(")");

        PreparedStatement ps = conn.prepareStatement(sql.toString());

        List<String> columns = Arrays.asList(tp.getSinkColumns().split(","));
        for (int i = 0; i < columns.size(); i++) {
//            String v = data.getString(columns.get(i)) == null ? null : data.getString(columns.get(i));
            ps.setString(i+1,data.getString(columns.get(i)));
        }

        ps.execute();
        conn.commit();
        ps.close();
        conn.close();
    }
}
